

package com.hazelcast.spi.impl.operationexecutor.impl;

import com.hazelcast.spi.impl.operationservice.PartitionTaskFactory;

import java.util.BitSet;

import static com.hazelcast.spi.impl.operationexecutor.impl.OperationExecutorImpl.getPartitionThreadId;

/**
 * A 'batch' of tasks to be executed on a partition thread.
 */
public class TaskBatch {

    private final PartitionTaskFactory taskFactory;
    private final BitSet partitions;
    private final int threadId;
    private final int partitionThreadCount;
    private int nextPartitionId;

    public TaskBatch(PartitionTaskFactory taskFactory, BitSet partitions, int threadId, int partitionThreadCount) {
        this.taskFactory = taskFactory;
        this.partitions = partitions;
        this.threadId = threadId;
        this.partitionThreadCount = partitionThreadCount;
    }

    public PartitionTaskFactory taskFactory() {
        return taskFactory;
    }

    /**
     * Gets the next task to execute.
     *
     * @return the task to execute, or null if the batch is complete.
     */
    public Object next() {
        int partitionId = nextPartitionId();
        return partitionId == -1 ? null : taskFactory.create(partitionId);
    }

    private int nextPartitionId() {
        for (; ; ) {
            int partitionId = partitions.nextSetBit(nextPartitionId);
            if (partitionId == -1) {
                return -1;
            }

            nextPartitionId = partitionId + 1;

            if (getPartitionThreadId(partitionId, partitionThreadCount) == threadId) {
                // only selected partitions that belong to the right partition thread.
                return partitionId;
            }
        }
    }
}
